package com.example.tys.baseapplication.net_tcp.frame.NettyImpl.udp;


import com.example.tys.baseapplication.net_tcp.frame.core.IClient;
import com.example.tys.baseapplication.net_tcp.frame.packet.BasePacket;

import java.net.InetSocketAddress;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeoutException;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramChannel;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;

/**
 * udp
 * 流程控制，在必要的位置预留回调接口
 * Created by tys on 2019/4/29.
 */

abstract class AbstractUdp<PacketRequest extends BasePacket, PacketResponse extends BasePacket>
        implements IClient<PacketRequest, PacketResponse> {

    /**
     * 绑定ip
     */
    private final static String HOST_NAME = "localhost";

    /**
     * 绑定端口
     */
    private final static int PORT = 1236;


    private NioEventLoopGroup mEventExecutors;
    private Bootstrap mUdpBootstrap;
    private final Channel mChannel;


    public AbstractUdp() {
        this.mEventExecutors = new NioEventLoopGroup();
        this.mUdpBootstrap = new Bootstrap()
                .group(mEventExecutors)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .handler(new ChannelInitializer<DatagramChannel>() {
                    @Override
                    protected void initChannel(DatagramChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("BaseHandle",new UdpBaseHandle());
                    }
                });
        this.mChannel = this.mUdpBootstrap.bind(new InetSocketAddress(getBindHostName(), getBindPort()))
                .syncUninterruptibly().channel();

    }

    public PacketResponse send(PacketRequest packet) throws TimeoutException, InterruptedException, RejectedExecutionException {
        return send(packet, -1);
    }


    @Override
    public PacketResponse send(PacketRequest packetRequest, int timeOutMillis) throws TimeoutException, InterruptedException, RejectedExecutionException {
        if (null == packetRequest) {
            throw new RejectedExecutionException("rejected send msg, msg is null");
        }
        if (beforeSend(packetRequest)) {
            throw new RejectedExecutionException("before sendSync, rejected sendSync msg " + (packetRequest.uniqueId
                    + " " + packetRequest.toString()));
        } else {
            channelSend(packetRequest);
            return afterSendAndWait(packetRequest, timeOutMillis);
        }
    }

    private void channelSend(PacketRequest packet) {
        this.mChannel.writeAndFlush(encoder(packet));
    }


    private class UdpBaseHandle extends SimpleChannelInboundHandler<DatagramPacket> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
            System.out.println("read:" + Thread.currentThread().getName());
            if (!isSelf(msg)) {
                PacketResponse packet = decoder(msg);
                afterReceive(packet);
            } else {
                fromSelfMsg(msg);
            }
        }

        private boolean isSelf(DatagramPacket msg) {
            InetSocketAddress s = msg.recipient();
            InetSocketAddress t = msg.sender();
            return s.equals(t);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("exceptionCaught");
            super.exceptionCaught(ctx, cause);
        }
    }

    /*======================================================================*/

    /**
     * 绑定ip
     *
     * @return
     */
    protected String getBindHostName() {
        return HOST_NAME;
    }

    /**
     * 绑定端口
     *
     * @return
     */
    protected int getBindPort() {
        return PORT;
    }

    /**
     * 发送前(网络判断 校验)
     *
     * @param packet
     * @return true 拦截,不再发送
     */
    protected boolean beforeSend(PacketRequest packet) {
        return false;
    }

    /**
     * 发送后
     *
     * @param packet
     * @param timeOutMillis
     */
    protected abstract PacketResponse afterSendAndWait(PacketRequest packet, int timeOutMillis) throws TimeoutException, InterruptedException,
            RejectedExecutionException;


    /**
     * 编码
     *
     * @param packet
     * @return
     */
    protected abstract DatagramPacket encoder(PacketRequest packet) ;


    /**
     * 解码
     *
     * @param packet
     * @return
     */
    protected abstract PacketResponse decoder(DatagramPacket packet);

    /**
     * 接收后(消息來自其他主机，非自身)
     *
     * @param packet
     */
    protected abstract void afterReceive(PacketResponse packet) ;

    /**
     * 接收到来自,自身的消息。不会调用{@link #decoder(DatagramPacket)} {@link #afterReceive(BasePacket)}
     *
     * @param msg
     */
    protected void fromSelfMsg(DatagramPacket msg) {
        byte[] out = new byte[msg.content().readableBytes()];
        msg.content().readBytes(out);
        System.out.println("fromSelfMsg:" + new String(out));
    }


    @Override
    public void close() {
        if (mEventExecutors != null) {
            mEventExecutors.shutdownGracefully()/*.sync()*/;
        }
        if (mUdpBootstrap != null) {
            mUdpBootstrap = null;
        }
    }


}
